package org.openinsula.arena.aws.integration.sqs;

import javax.annotation.Resource;

import org.openinsula.arena.aws.AwsConstants;
import org.openinsula.arena.aws.sqs.SqsManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageSource;
import org.springframework.util.Assert;

import com.xerox.amazonws.sqs2.SQSException;

/**
 * {@link MessageSource} implementation that delegates to a {@link SqsManager}
 * to poll a SQS queue. Each poll of the queue returns only one message.
 * 
 * <p>
 * The SQS queue name is provided in the {@link AwsConstants#SQS_QUEUE_NAME}
 * header.
 * 
 * @author Edson Yanaga
 */
public class SqsReceivingMessageSource implements MessageSource<String>, InitializingBean {

	protected final Logger logger = LoggerFactory.getLogger(getClass());

	@Resource(name = "sqsManager")
	private SqsManager sqsManager;

	private String queueName;

	@Override
	public void afterPropertiesSet() throws Exception {
		Assert.notNull(sqsManager, "'sqsManager' must not be null");
		Assert.notNull(queueName, "'queueName' must not be null");
	}

	@Override
	public Message<String> receive() {
		try {
			com.xerox.amazonws.sqs2.Message message = sqsManager.get(queueName);
			if (message != null) {
				String value = message.getMessageBody();
				if (value != null) {
					logger.debug("received SQS message [%s]", value);
					Message<String> result = MessageBuilder.withPayload(value).setHeader(AwsConstants.SQS_QUEUE_NAME,
							queueName).build();

					sqsManager.delete(queueName, message);

					return result;
				}
			}
		}
		catch (SQSException e) {
			throw new MessagingException("failure occurred while polling for SQS message", e);
		}
		return null;
	}

	public void setSqsManager(SqsManager sqsManager) {
		this.sqsManager = sqsManager;
	}

	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

}
